package org.apache.hadoop.hdfs.server.datanode;

import java.io.BufferedInputStream;
import java.io.Closeable;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.FSConstants;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.DataChecksum;

/**
 * New class introduced to actually read the block in a cyclical manner, passing it on to the 
 * BlockSenders which actually send the data back.
 * 
 * @author hadoopuser
 *
 */
public class BlockReader extends Thread implements FSConstants, Closeable {
	

	private boolean debugMode = false;
	
	public static final Log LOG = DataNode.LOG;
	
	private List<BlockSender> registeredSenders = new ArrayList<BlockSender>();
	private Map<BlockSender,LatchedArrayBlockingQueue<Chunk>> chunkQueues = new HashMap<BlockSender,LatchedArrayBlockingQueue<Chunk>>(); 
	
	private DataNode datanode;
	private Block block;
	private long blockLength;
	private InputStream blockIn; // block input data stream
	
	private DataChecksum checksum; // checksum stream
	private DataInputStream checksumIn; // checksum input datastream
	private int bytesPerChecksum; // chunk size
	private int checksumSize; // checksum size
	private boolean terminate = false;
	
	public int getBytesPerCheckSum() {
		return bytesPerChecksum;
	}

	
	public long getBlockLength() {
		return blockLength;
	}
	
	public int getChecksumSize() {
		return checksumSize;
	}
	
	public BlockReader(Block block, DataNode datanode, boolean corruptChecksumOK) throws IOException {
		
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Creating new BlockReader for Block " + block.getBlockId()); } 
		
		this.block = block;
		this.datanode = datanode;
		
    	try {


    		this.blockLength = datanode.data.getLength(block);
    		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Block " + block.getBlockId() + " Length determined as: " + this.blockLength); }

    		if ( datanode.data.metaFileExists(block) ) { // Checksumming always done, original code included check on checksum being selected
    			checksumIn = new DataInputStream(new BufferedInputStream(this.datanode.data.getMetaDataInputStream(block), BUFFER_SIZE));

    			// read and handle the common header here. For now just a version
    			BlockMetadataHeader header = BlockMetadataHeader.readHeader(checksumIn);
    			short version = header.getVersion();
    			
    			

    			if (version != FSDataset.METADATA_VERSION) {
    				LOG.warn("Wrong version (" + version + ") for metadata file for " + block + " ignoring ...");
    			}
    			checksum = header.getChecksum();
    		} 
    		else
    		{
    			LOG.warn("Could not find metadata file for " + block);
    			// This only decides the buffer size. Use BUFFER_SIZE?
    			checksum = DataChecksum.newDataChecksum(DataChecksum.CHECKSUM_NULL, 16 * 1024);
    		}
      
    		bytesPerChecksum = checksum.getBytesPerChecksum();
    		if (bytesPerChecksum > 10*1024*1024 && bytesPerChecksum > blockLength) {
    			checksum = DataChecksum.newDataChecksum(checksum.getChecksumType(), Math.max((int)blockLength, 10*1024*1024));
    			bytesPerChecksum = checksum.getBytesPerChecksum();        
    		}
    		checksumSize = checksum.getChecksumSize();
    		
    		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Bytes per checksum: " + bytesPerChecksum); }
    		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Checksum size: " + checksumSize); }
    		
    		//blockIn = datanode.data.getBlockInputStream(block, 0); // seek to offset Not needed here, done in run method.


    	} catch (IOException ioe) {
    		
    		IOUtils.closeStream(this);
    		IOUtils.closeStream(checksumIn);
    		ioe.printStackTrace();
    		throw ioe;
    	}
	    
		
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Finished BlockReader constructor for block " + block.getBlockId()); }
	}
	
	public DataChecksum getChecksum() {
		return this.checksum;
	}
	
	public boolean equals(Object object) {
		if (object instanceof BlockReader) {
			BlockReader blockReader = (BlockReader) object;
			return blockReader.block.getBlockId() == this.block.getBlockId();
		}
		
		return false;
	}
	
	public int hashCode() {
		return this.block.hashCode();
	}
	
	public synchronized void register(BlockSender sender) throws IOException {
		
		if (PrototypeUtilities.SUMMARY_DEBUG_MODE) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Registering sender: " + sender.getSenderID() + " with BlockReader for Block " + block.getBlockId()); }
		int registeredSendersCount = 0;
		synchronized (registeredSenders) {
			registeredSenders.add(sender);
			registeredSendersCount = registeredSenders.size();
		}
		
		if (PrototypeUtilities.SUMMARY_DEBUG_MODE) { 
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Sender: " + sender.getSenderID() + " now registered"); 
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " There are now: " + registeredSendersCount + " senders registered");
		}
		
		
	}
	
	public synchronized LatchedArrayBlockingQueue<Chunk> activate(BlockSender sender) {
		
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Activating sender: ID " + sender.getSenderID()); }
		
		LatchedArrayBlockingQueue<Chunk> queueForSenderToPullChunksFrom = null;
		
		synchronized (registeredSenders) {
			if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Got registeredSenders lock A"); }
			synchronized (chunkQueues) {
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Got chunkQueus lock A"); }
				if (registeredSenders.contains(sender)) {
					// TODO Account for the maximum number of chunks in a file when making the queue length - no point having a queue that is larger than it.
					queueForSenderToPullChunksFrom = new LatchedArrayBlockingQueue<Chunk>(PrototypeUtilities.MAX_LENGTH_OF_CHUNK_QUEUES);
					chunkQueues.put(sender, queueForSenderToPullChunksFrom);
					registeredSenders.remove(sender);
				}
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Sending notify all to chunkQueues"); }
				chunkQueues.notifyAll();
			}
			if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Released chunkQueues lock A"); }
			
		}
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Released registeredSenders lock A"); }
		
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Finished Activating sender: ID" + sender.getSenderID()); }
		
		return queueForSenderToPullChunksFrom;
	}
	
	/**
	 * 
	 * @param sender
	 * @return True if there are no registered senders left (i.e. is empty)
	 */
	public synchronized boolean deregister(BlockSender sender) {
		
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Deregistering sender " + sender.getSenderID()); }
		
		synchronized (registeredSenders) {
			if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Got registeredSenders lock B"); }
			synchronized (chunkQueues) {
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Got chunkQueus lock B"); }
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Got locks. Now actually deregistering sender " + sender.getSenderID()); }
				registeredSenders.remove(sender);
				chunkQueues.remove(sender);
				if (registeredSenders.isEmpty() && chunkQueues.isEmpty()) {
					this.terminate = true;
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Done deregistering sender (terminate set)"); }
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Released registeredSenders/chunkQueues lock B"); }
					return true;
				}
				else {
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Done deregistering sender (terminate NOT set)"); }
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Released registeredSenders/chunkQueues lock B"); }
					return false;
				}

			}
			
			
		}
		
		
		
		
	}


	@Override
	public void run() {
		
	   int maxChunksPerPacket = Math.max(1,(BUFFER_SIZE + bytesPerChecksum - 1)/bytesPerChecksum);
	   int pktSize = DataNode.PKT_HEADER_LEN + SIZE_OF_INTEGER;
       pktSize += (bytesPerChecksum + checksumSize) * maxChunksPerPacket;
       
       
		
		if (debugMode) { 
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " BlockReader starting run method");
			
			System.out.println("BlockReader made calculation of maxChunksPerPacket: " + maxChunksPerPacket + " and pktSize " + pktSize);
			
		}		
		
		while (!terminate) {
			
			
			// Reset the streams back to start position
			try {
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Resetting block input stream to position 0"); }
				blockIn = datanode.data.getBlockInputStream(block, 0); // seek to offset
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Resetting checksum input stream to position " + BlockMetadataHeader.getHeaderSize()); }
				checksumIn = new DataInputStream(new BufferedInputStream(this.datanode.data.getMetaDataInputStream(block), BUFFER_SIZE));
				IOUtils.skipFully(checksumIn, BlockMetadataHeader.getHeaderSize());
			}
			catch (IOException ex) {
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " IOException after attempting to reset streams: " + ex.getMessage()); }
				ex.printStackTrace();
				terminate = true;
				break;
			}
			
			// Don't bother if there are no queues to distribute the data to.
			synchronized (chunkQueues) {
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Got chunkQueues lock C"); }
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + this); }
				while (chunkQueues.isEmpty()) {
					try {
						if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Waiting until there are queues to receive chunks"); }
						chunkQueues.wait();
						if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Woken - need to check again if there are queues to receive chunks"); }
					} catch (InterruptedException e) {
						System.err.println(e.getMessage() + " (Interrupted Exception)");
						continue;
					}
				}
			}
			if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Released chunkQueues lock C"); }
			
			if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " There are now queues to receive chunks"); }
			
			long currentPositionInFile = 0;
			while (currentPositionInFile < this.blockLength) {
				
				// Read in block, in order to create chunk
				
				long readFrom = currentPositionInFile;
				long readTo = Math.min(readFrom + (maxChunksPerPacket * bytesPerChecksum), this.blockLength) - 1;
				int bytesRead = (int) (readTo - readFrom) + 1;
				long chunksRead = bytesRead / bytesPerChecksum;
//				
				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Attempting to read in block from: " + readFrom + " to " + readTo); }
//				
				byte[] chunkData = new byte[bytesRead];
				byte[] chunkChecksum = new byte[(int) (this.checksumSize * chunksRead)];

				if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Created byte array of length: " + chunkData.length); }
				
				try {
					IOUtils.readFully(blockIn, chunkData, 0, bytesRead);
					IOUtils.readFully(checksumIn, chunkChecksum, 0, (int) chunksRead * checksumSize);

					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Read in block from " + readFrom + " to " + readTo); }
				
					currentPositionInFile = ++readTo;
					
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Current position in file now set to: " + currentPositionInFile); }
					// IS it possible to verify checksum here before sending the data onwards?
				}
				catch (IOException ex) {
					terminate = true;
					ex.printStackTrace();
					break;
					
				}
				
				for (int i = 0; i < chunksRead; i++) {
					
					// Create chunk of data
					long chunkStart = readFrom + (bytesPerChecksum * i) ;
					
					Chunk chunk = new Chunk(block, Arrays.copyOfRange(chunkData, i * bytesPerChecksum, (i * bytesPerChecksum) + bytesPerChecksum), chunkStart, (chunkStart + bytesPerChecksum - 1), Arrays.copyOfRange(chunkChecksum, i * checksumSize, i * checksumSize + checksumSize));
					
					// 	Chunk chunk = new Chunk(block, chunkData, readFrom, readTo, chunkChecksum);
					
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Distributing chunk to queues. Block: " + block.getBlockId()); }
					
					synchronized (chunkQueues) {
						
						// Distribute it to each active listener's queue.
						if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Got chunkQueues lock D"); }
						for (BlockSender blockSender : chunkQueues.keySet()) {
							
							if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Distributing chunk to queue for sender " + blockSender.getSenderID()); }
							
							while (true) {
								
								BlockingQueue<Chunk> queue = chunkQueues.get(blockSender);
								try {
									if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Remaining capacity for queue: " + blockSender.getSenderID() + " is " + queue.remainingCapacity() + " (BlockReader: " + block.getBlockId()); }
									if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Attempt to put chunk onto queue (BlkReader:" + block.getBlockId() + "Sender: "+ blockSender.getSenderID() + "Chunk start: "+ chunk.getStartOffset() + " Chunk end: "+ chunk.getEndOffset()); }
									queue.put(chunk);
									if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Chunk placed on queue for sender " + blockSender.getSenderID()); }
									break;
								}
								catch (InterruptedException ex) {
									if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Interrupted Exception " + ex.getMessage()); }
									continue;
								}

							}
						}
					}
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + "Released chunkQueues lock D"); }
				}
				
				
				if (terminate || chunkQueues.isEmpty()) {
					if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " BlockReader terminating"); }
					break;
				}
			}
			
			try {
				blockIn.close();
				checksumIn.close();
			}
			catch (Exception ex) {
				ex.printStackTrace();
			}

			
		}
		
		if (debugMode) { System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " BlockReader ending run method"); }
		
	}
	
	public boolean blockInInstanceOfFileInputStream() {
		return (blockIn instanceof FileInputStream);
	}

	@Override
	public void close() throws IOException {
		
	    IOException ioe = null;
	    // close checksum file
	    if(checksumIn!=null) {
	      try {
	        checksumIn.close();
	      } catch (IOException e) {
	        ioe = e;
	        e.printStackTrace();
	      }
	      checksumIn = null;
	    }
	    // close data file
	    if(blockIn!=null) {
	      try {
	        blockIn.close();
	      } catch (IOException e) {
	        ioe = e;
	        e.printStackTrace();
	      }
	      blockIn = null;
	    }
	    // throw IOException if there is any
	    if(ioe!= null) {
	      throw ioe;
	    }
		
	}
	
	public String toString() {
		StringBuilder builder = new StringBuilder();
		
		builder.append("BlockReader for Block " + block.getBlockId() + "\n");
		synchronized (chunkQueues) {
			builder.append("Active Queues:");
			for (BlockSender sender : chunkQueues.keySet()) {
				builder.append(" " + sender.getSenderID() + ":" + chunkQueues.get(sender).size());
			}
		}
		builder.append("\n");
		
		return builder.toString();
	}
}